package com.bw.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.bw.gmall.realtime.utils.DateFormatUtil;
import com.bw.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

//日志数据  分流
public class BaseLogApp {
    public static void main(String[] args) throws Exception {
        //初始化
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //从kafka读取数据
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("topic_log", "base_log_app"));
        //数据清洗
        OutputTag<String> dirtyStream = new OutputTag<String>("dirtyStream"){};
        SingleOutputStreamOperator<String> clearDs = source.process(new ProcessFunction<String, String>() {

            @Override
            public void processElement(String value, ProcessFunction<String, String>.Context ctx, Collector<String> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    out.collect(jsonObject.toJSONString());
                } catch (Exception e) {
                    ctx.output(dirtyStream, value);
                }
            }
        });
//        clearDs.print();
        //转换主流数据结构
        SingleOutputStreamOperator<JSONObject> MapDs = clearDs.map(JSON::parseObject);

        //添加一级渠道 二级渠道
        SingleOutputStreamOperator<JSONObject> enrichedLogDs = MapDs.map(log -> {
            JSONObject common = log.getJSONObject("common");
            if (common != null) {
                // 提取页面信息或来源信息（根据实际业务字段判断）
                String page_id = log.getJSONObject("page") != null ?
                        log.getJSONObject("page").getString("page_id") : null;
                String ch = log.getJSONObject("common") != null ?
                        log.getJSONObject("common").getString("ch") : null;



                // 根据页面ID或来源字段判断一级来源和二级来源
                String firstSource = "未知来源";
                String secondSource = "未知页面";

                if (ch != null) {
                    switch (ch) {
                        case "oppo":
                            firstSource = "店铺流量";
                            secondSource = "OPPO";
                            break;
                        case "xiaomi":
                            firstSource = "店铺流量";
                            secondSource = "小米";
                            break;
                        case "huawei":
                            firstSource = "店铺流量";
                           secondSource = "华为";
                            break;
                        case "360":
                            firstSource = "店铺流量";
                            secondSource = "360";
                            break;
                        case "vivo":
                            firstSource = "店铺流量";
                            secondSource = "VIVO";
                            break;
                        case "Appstore":
                            firstSource = "商品流量";
                            if(page_id!=null){
                                switch (page_id) {
                                    case "good_list":
                                        secondSource = "商品列表";
                                        break;
                                    case "good_detail":
                                        secondSource = "商品详情页";
                                        break;
                                    case "cart":
                                        secondSource = "购物车";
                                        break;
                                    case "order":
                                        secondSource = "订单页";
                                        break;
                                    case "pay":
                                        secondSource = "支付页";
                                        break;
                                    default:
                                        secondSource = "未知页面";
                                        break;
                                }
                            }
                            break;
                        case "wandoujia":
                            firstSource = "直播流量";
                             if(page_id!=null){
                            switch (page_id) {
                                case "good_list":
                                    secondSource = "商品列表";
                                    break;
                                case "good_detail":
                                    secondSource = "商品详情页";
                                    break;
                                case "cart":
                                    secondSource = "购物车";
                                    break;
                                case "order":
                                    secondSource = "订单页";
                                    break;
                                case "pay":
                                    secondSource = "支付页";
                                    break;
                                default:
                                    secondSource = "未知页面";
                                    break;
                            }
                        }
                            break;
                        case "web":
                            firstSource = "内容流量";
                            if(page_id!=null){
                                switch (page_id) {
                                    case "good_list":
                                        secondSource = "商品列表";
                                        break;
                                    case "good_detail":
                                        secondSource = "商品详情页";
                                        break;
                                    case "cart":
                                        secondSource = "购物车";
                                        break;
                                    case "order":
                                        secondSource = "订单页";
                                        break;
                                    case "pay":
                                        secondSource = "支付页";
                                        break;
                                    default:
                                        secondSource = "未知页面";
                                        break;
                                }
                            }
                                break;
                        default:
                            // 可扩展其他页面类型
                            break;
                    }
                }

                // 添加到common字段中
                common.put("first_source", firstSource);
                common.put("second_source", secondSource);
            }
            return log;
        });


        //新老用户校验
        //根据mid分组
        KeyedStream<JSONObject, String> midDs = enrichedLogDs.keyBy(json -> json.getJSONObject("common").getString("mid"));
        //状态编程
        SingleOutputStreamOperator<JSONObject> isNewDs = midDs.process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {
            ValueState<String> state;

            @Override
            public void open(Configuration parameters) throws Exception {
                super.open(parameters);
                state = getRuntimeContext().getState(new ValueStateDescriptor<>("state", String.class));
            }

            @Override
            public void processElement(JSONObject value, KeyedProcessFunction<String, JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                //获取状态
                String isNew = value.getJSONObject("common").getString("is_new");
                //获取状态时间
                String stateTime = state.value();
                //获取数据时间
                Long dataTime = value.getLong("ts");
                //转换时间格式
                DateFormatUtil.toDate(dataTime);

                //判断
                if ("1".equals(isNew)) {
                    if (stateTime == null) {
                        out.collect(value);
                    } else {
                        if (!stateTime.equals(DateFormatUtil.toDate(dataTime))) {
                            isNew = "0";
                            value.getJSONObject("common").put("is_new", isNew);
                        }
                    }
                } else {
                    if (stateTime == null) {
                        String yesterday = DateFormatUtil.toDate(dataTime - 24 * 60 * 60 * 1000L);
                        state.update(yesterday);
                    }
                }
            }
        });
//        isNewDs.print();
        //分流
        OutputTag<String> startTag = new OutputTag<String>("startT"){};
        OutputTag<String> displayTag = new OutputTag<String>("displayT"){};
        OutputTag<String> actionTag = new OutputTag<String>("actionT"){};
        OutputTag<String> errorTag = new OutputTag<String>("errorT"){};
        SingleOutputStreamOperator<String> separatedStream = isNewDs.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject value, ProcessFunction<JSONObject, String>.Context ctx, Collector<String> out) {
                //收集错误数据
                JSONObject err = value.getJSONObject("err");
                if (err != null) {
                    ctx.output(errorTag, value.toJSONString());
                }
                //删除‘err’字段
                value.remove("err");

                //收集启动数据
                JSONObject start = value.getJSONObject("start");
                if (start != null) {
                    ctx.output(startTag, value.toJSONString());
                }else {
                         //获取page字段
                         JSONObject page = value.getJSONObject("page");
                         //获取common字段
                         JSONObject common = value.getJSONObject("common");
                         //获取ts字段
                         Long ts = value.getLong("ts");
                        //收集曝光数据
                        JSONArray displays = value.getJSONArray("displays");
                        if (displays != null) {
                            for (int i = 0; i < displays.size(); i++) {
                                JSONObject display = displays.getJSONObject(i);
                                //创建空json对象
                                JSONObject displayObj = new JSONObject();
                                displayObj.put("display", display);
                                displayObj.put("common", common);
                                displayObj.put("page", page);
                                displayObj.put("ts", ts);
                                ctx.output(displayTag, displayObj.toJSONString());

                            }
                        }
                        //收集动作数据
                        JSONArray actions = value.getJSONArray("actions");
                        if (actions != null) {
                            for (int i = 0; i < actions.size(); i++) {
                                JSONObject action = actions.getJSONObject(i);
                                //创建空json对象
                                JSONObject actionObj = new JSONObject();
                                actionObj.put("action", action);
                                actionObj.put("common", common);
                                actionObj.put("page", page);
                                actionObj.put("ts", ts);
                                ctx.output(actionTag, actionObj.toJSONString());
                            }
                        }

                    value.remove("displays");
                    value.remove("actions");
                    // 收集页面数据
                    out.collect(value.toJSONString());
                }
            }
        });

        // 打印主流和各侧输出流查看分流效果
        separatedStream.print("page------");
//        separatedStream.getSideOutput(startTag).print("start-------");
//        separatedStream.getSideOutput(displayTag).print("display--------");
//        separatedStream.getSideOutput(actionTag).print("action-------");
//        separatedStream.getSideOutput(errorTag).print("error-------");

       //  提取各侧输出流
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errorTag);
       //  定义不同日志输出到 Kafka 的主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";


        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));
        //启动
        env.execute();

    }
}
